package com.jack.code;


import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.Arrays;
import java.util.Date;

/**
 * Created by zhang_j on 2019/10/21
 */
public class DataMonitor implements Watcher, StatCallback {
    private ZooKeeper zk;
    private String znode;
    private Watcher chainedWatcher;
    boolean dead;
    private DataMonitorListener listener;
    private byte[] prevData;

    public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher, DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.chainedWatcher = chainedWatcher;
        this.listener = listener;
        // Get things started by checking if the node exists. We are going
        // to be completely event driven
        zk.exists(znode, true, this, null);
    }

    public void processResult(int rc, String path, Object ctx, Stat stat) {
        boolean exists;

        System.out.println("STAT CALL BACK");
        System.out.println("rc " + rc);
        System.out.println("path " + path);
        System.out.println("ctx " + ctx);
        System.out.println("stat " + stat);

        switch (Code.get(rc)){
            case OK:
                exists = true;
                break;
            case NONODE:
                exists = false;
                break;
            case SESSIONEXPIRED:
            case NOAUTH:
                dead = true;
                listener.closing(rc);
                return;
            default:
                // Retry errors
                zk.exists(znode, true, this, null);
                return;
        }

        byte[] b = null;
        if(exists){
            try{
                b = zk.getData(znode,false,null);
            }catch (KeeperException e){
                e.printStackTrace();
            }catch (InterruptedException e){
                return;
            }
            System.out.println("GET DATA OF " + new String(b));
            System.out.println("GET DATA END ");
        }

        if(b == null && b != prevData || (b!=null && !Arrays.equals(prevData,b))){
            listener.exists(b);
            prevData = b;
        }
    }

    public void process(WatchedEvent event) {
        String path = event.getPath();
        System.out.println("GOT EVENT " + event + " @" + new Date() + ",type is " + event.getType());

        if (event.getType() == Event.EventType.None) {
            // We are are being told that the state of the
            // connection has changed
            switch (event.getState()) {
                case SyncConnected:
                    // In this particular example we don't need to do anything
                    // here - watches are automatically re-registered with
                    // server and any watches triggered while the client was
                    // disconnected will be delivered (in order of course)
                    break;
                case Expired:
                    // It's all over
                    dead = true;
                    listener.closing(Code.SESSIONEXPIRED.intValue());
                    break;
                case AuthFailed:
                    break;
                case ConnectedReadOnly:
                    break;
                case Disconnected:
                    break;
                case SaslAuthenticated:
                    break;
                default:
                    break;
            }
        } else {
            if (path != null && path.equals(znode)){
                zk.exists(znode,true,this,null);
            }
        }

        if(chainedWatcher != null){
            chainedWatcher.process(event);
        }
    }

}
